package com.inns.position.hbase.util;

import com.inns.position.hbase.HbaseConsts;
import com.inns.position.hbase.StatType;
import com.inns.position.hbase.bean.StatisticsResult;
import com.inns.position.hbase.config.properties.InnsProperties;
import com.inns.position.hbase.coprocessor.service.ObserverStatisticsProtos;
import com.xxdai.ace.persistence.hbase.jdbc.coprocessor.service.ExpandAggregationProtos;
import com.xxdai.ace.persistence.hbase.jdbc.exception.AssertPropertiseException;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.*;

public class StatisticsUtil {

    public static String getRowKey(StatType statType, String middle, long endTime) {
        if (!middle.endsWith("|")) {
            middle = middle + "|";
        }
        return statType.getType() + "|" + middle + endTime / 1000;
    }

    public static List<StatisticsResult> doStatistics(RegionCoprocessorEnvironment env, ObserverStatisticsProtos.Request request) {
        List<StatisticsResult> resultList = new ArrayList<>();
        Map<String, Integer> indexMap = new HashMap<>();
        InternalScanner scanner = null;
        try {
            scanner = env.getRegion().getScanner(new Scan());
            boolean hasMoreRows;
            List<Cell> rowCells = new ArrayList<>();
            do {
                hasMoreRows = scanner.next(rowCells);
                StatisticsUtil.dealRowCells(rowCells, request, true, resultList, indexMap);
                rowCells.clear();
            } while (hasMoreRows);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (scanner != null) {
                try {
                    scanner.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        return resultList;
    }

    public static void printResultList(List<StatisticsResult> resultList) {
        for (StatisticsResult map : resultList) {
            System.out.println(map);
        }
        System.out.println("总记录数：" + resultList.size());
    }

    public static void dealRowCells(List<Cell> rowCells, ObserverStatisticsProtos.Request request, boolean needSumTime, List<StatisticsResult> resultList, Map<String, Integer> indexMap) {
        List<ObserverStatisticsProtos.ExpandCell> groupColumnsList = request.getGroupColumnsList();
        List<ObserverStatisticsProtos.ExpandCell> countColumnsList = request.getCountColumnsList();

        Map<String, String> groupKeys = new HashMap<>();
        Map<String, String> countValues = new HashMap<>();
        String startTime = null;
        String endTime = null;
        continueTag:
        for (Cell cell : rowCells) {
            String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
            String qualify = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            for (ObserverStatisticsProtos.ExpandCell gCell : groupColumnsList) {
                if (family.equals(gCell.getFamily().toStringUtf8()) && qualify.equals(gCell.getQualify().toStringUtf8())) {
                    String groupValue = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    groupKeys.put(gCell.getQualify().toStringUtf8(), groupValue);
                    continue continueTag;
                }
            }
            for (ObserverStatisticsProtos.ExpandCell cCell : countColumnsList) {
                if (family.equals(cCell.getFamily().toStringUtf8()) && qualify.equals(cCell.getQualify().toStringUtf8())) {
                    String countValue = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    countValues.put(cCell.getQualify().toStringUtf8(), countValue);
                    continue continueTag;
                }
            }
            if (needSumTime) {
                if (family.equals("info") && qualify.equals("startTime")) {
                    startTime = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                } else if (family.equals("info") && qualify.equals("endTime")) {
                    endTime = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                }
            }
        }
        long sumTime = 0;
        if (startTime != null && endTime != null) {
            sumTime = Long.parseLong(endTime) - Long.parseLong(startTime);
            if (sumTime < HbaseConsts.MIN_STAT_TIME()) {
                sumTime = 0;
            }
        }
        String groupKey = "";
        for (ObserverStatisticsProtos.ExpandCell gCell : groupColumnsList) {
            groupKey = groupKey + groupKeys.get(gCell.getQualify().toStringUtf8()) + "|";
        }

        Integer index = indexMap.get(groupKey);
        StatisticsResult groupRowMap;
        if (index == null) {
            groupRowMap = new StatisticsResult();
            groupRowMap.setGroupKey(groupKey);
            indexMap.put(groupKey, indexMap.size());
            resultList.add(resultList.size(), groupRowMap);

            groupRowMap.setCount(1);
            groupRowMap.setSum(sumTime);

            Map<String, Set<String>> distinctValues = new HashMap<>();
            Map<String, Integer> countMap = new HashMap<>();
            dealCount(countValues, distinctValues, countMap);

            groupRowMap.setDistinctValues(distinctValues);
            groupRowMap.setCountMap(countMap);
        } else {
            groupRowMap = resultList.get(index);
            groupRowMap.increaseCount();
            groupRowMap.addSum(sumTime);

            dealCount(countValues, groupRowMap.getDistinctValues(), groupRowMap.getCountMap());
        }
    }

    private static void dealCount(Map<String, String> countValues, Map<String, Set<String>> distinctValues, Map<String, Integer> countMap) {
        if (countValues == null || countValues.size() == 0) {
            return;
        }
        if (distinctValues == null) {
            return;
        }
        countMap.clear();
        if (distinctValues.size() == 0) {
            for (Map.Entry<String, String> entry : countValues.entrySet()) {
                distinctValues.put(entry.getKey(), new HashSet<>(Arrays.asList(entry.getValue())));
                countMap.put(entry.getKey(), 1);
            }
        } else {
            for (Map.Entry<String, String> entry : countValues.entrySet()) {
                distinctValues.get(entry.getKey()).add(entry.getValue());
                countMap.put(entry.getKey(), distinctValues.get(entry.getKey()).size());
            }
        }
    }

    public static ObserverStatisticsProtos.ExpandCell toExpandCell(ExpandAggregationProtos.ExpandCell cell) {
        if (cell == null) return null;
        AssertPropertiseException.notNull(cell.getFamily(), "列簇不能为空！");
        AssertPropertiseException.notNull(cell.getQualify(), "列不能为空！");
        ObserverStatisticsProtos.ExpandCell.Builder builder = ObserverStatisticsProtos.ExpandCell.newBuilder();
        builder.setFamily(cell.getFamily());
        builder.setQualify(cell.getQualify());
        if (cell.getValue() != null)
            builder.setValue(cell.getValue());
        return builder.build();
    }

    public static List<StatisticsResult> getGroupAndCount(ResultScanner scanner, List<ObserverStatisticsProtos.ExpandCell> groupCells, List<ObserverStatisticsProtos.ExpandCell> countCells) {
        List<StatisticsResult> resultList = new ArrayList<>();
        Map<String, Integer> indexMap = new HashMap<>();

        ObserverStatisticsProtos.Request.Builder builder = ObserverStatisticsProtos.Request.newBuilder();
        for (ObserverStatisticsProtos.ExpandCell groupCell : groupCells) {
            builder.addGroupColumns(groupCell);
        }
        for (ObserverStatisticsProtos.ExpandCell countCell : countCells) {
            builder.addCountColumns(countCell);
        }

        try {
            for (Result result : scanner) {
                List<Cell> rowCells = result.listCells();
                StatisticsUtil.dealRowCells(rowCells, builder.build(), true, resultList, indexMap);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                scanner.close();
            } catch (Exception ignored) {
            }
        }
        return resultList;
    }

    public static void fillResultMap(List<Cell> rowCells, Map<String, String> resultMap, Map<String, String> keyToKey) {
        for (Cell cell : rowCells) {
            String qualify = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            String rKey = keyToKey.get(qualify);
            if (rKey != null) {
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                resultMap.put(rKey, value);
            }
        }
    }
}
